package org.yray.io.rocketmqdemo.produce;


import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.yray.io.rocketmqdemo.listener.BaseMqMsg;
import org.yray.io.rocketmqdemo.listener.Notice;

@Service
@Slf4j
public class MqProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    //出异常了和消息状态非SEND_OK都要处理下,普通消息,以服务名作为topic,以业务名作为tags
    public void sendMsgToMq(String msgStr)
    {
        BaseMqMsg baseMqMsg = new BaseMqMsg(new Notice(msgStr));
        Message msg = new Message("MQ_TOPIC_NOTICE","noticeMsgListener", JSON.toJSONString(baseMqMsg).getBytes());

        try {
            SendResult result=rocketMQTemplate.getProducer().send(msg);
            log.info("send mq result={}",result.toString());
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    //延时消息
    public void sendDelayMsgToMq(String msgStr)
    {
        Message msg = new Message("my-topic","my-delay",msgStr.getBytes());
        msg.setDelayTimeLevel(2);//级别2对应5秒
        try {
            SendResult result=rocketMQTemplate.getProducer().send(msg);
            log.info("send mq result={}",result.toString());
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
